…3913)

* Send each logical database as its own independent schema snapshot

Each database now gets a fresh collection_started_at timestamp and its
own terminal payload with collection_payloads_count, so the backend can
independently complete or discard each database's snapshot. Errors on
one database are isolated: partial rows are discarded and collection
continues for the remaining databases.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* Add changelog entry for PR #23913

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* Fix metric bookkeeping for mid-iteration database failures

When a database fails after already flushing one or more chunks, two
counters were wrong:
- _total_rows_count included unsent buffered rows that were about to
  be discarded, inflating the tables_count gauge
- total_payloads_count excluded chunk payloads already emitted for
  the failed database, undercounting the payloads_count gauge

Fix by computing unsent row count before the delete and subtracting it,
and adding _collection_payloads_count to total_payloads_count in the
error path as well as the success path.

Add test for mid-iteration failure after a chunk flush.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* Simplify per-DB state reset using per-DB queued_rows initialization

Reset _queued_rows per database instead of tracking a slice offset.
On error, discard by reassignment rather than del slicing.
Use a local db_rows_count so _total_rows_count only reflects completed
databases, not partially-failed ones.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* Add row count to per-database collection debug log

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* Add payload count to per-database collection debug log

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* Fix changelog entry type: changed -> fixed

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* Move completion debug log after terminal flush so payload count is accurate

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* Include database info in terminal payload for empty databases

When a logical database has no rows (or all rows were already flushed
by chunk-size), the terminal payload would have metadata: [] with no
DatabaseName for the backend to apply per-database snapshot
completion/deletion to. For empty databases, append the database-level
dict as a sentinel row so the backend receives a DatabaseName even when
no schema rows were collected.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* Remove empty-database sentinel from per-DB schema collection

Drop the sentinel row that was injected into terminal payloads for
empty or fully-chunk-flushed databases. Per Seth's review, this is a
separate concern and should be addressed in a follow-up PR to keep
revert scope minimal.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* Guard against missing 'name' key in database entries from _get_databases()

`database['name']` raised KeyError if a subclass returned a malformed
entry, propagating to the outer handler and aborting the entire run.
Switch to `database.get('name')` and reorder so the guard check fires
before the debug log. Add a test covering the per-DB skip path.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* Include DB identity sentinel in terminal payload for empty databases

When a logical database has no rows, the terminal flush emits
metadata=[] with no DatabaseName. The backend cannot attribute
snapshot completion/deletion to that DB, so stale schemas survive.

Append a {**database} sentinel entry before the terminal flush when
db_rows_count==0. This ensures the backend calls snapshotStore.Add
with a DatabaseName and can remove stale rows on completion.

Update the test to assert the sentinel name field rather than empty
metadata, per Ilia's review request.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* Fix ClickHouse schema collection compatibility with empty-DB sentinel

ClickhouseSchemaCollector uses a single cluster-wide stub entry rather
than iterating real database names. Without this fix, the base-class
sentinel appends the stub name ('_cluster_') to empty terminal payloads,
sending a fake database identifier to the backend and breaking the
test_collect_emits_empty_snapshot_marker_when_no_tables assertion.

Add _emit_empty_db_sentinel class attribute (default True) to
SchemaCollector so subclasses using stub entries can opt out, and set it
to False in ClickhouseSchemaCollector.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* Revert empty-DB sentinel from SchemaCollector

The sentinel approach requires touching every integration that uses a
stub database entry (e.g. ClickHouse), making it unsuitable for this PR.
Deferred to a follow-up with proper cross-integration design.

Restores metadata=[] for empty-database terminal payloads.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* Fix ClickHouse test_main_query_failure_closes_client for per-DB error isolation

The base SchemaCollector now catches per-DB errors and continues (matching
the production behavior where run_job already catches all exceptions from
collect_schemas). Update the test to verify client cleanup without asserting
exception propagation, which was never a code-level contract.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

---------

Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>